#!/bin/bash

SUDO="sudo "
if [[ $(id -u) -eq 0 ]]; then SUDO=""; fi

read -p "Enter kafka version (default 3.7.0 e.g. 3.6.0 | 3.7.0 .etc): " version </dev/tty
version=$(echo "$version" | tr -s '[:space:]')
if [[ "$version" == "" ]]; then version="3.7.0"; fi

$SUDO mkdir -p /opt/kafka/${version}

if [[ -x /opt/kafka/$version/bin/kafka-server-start.sh ]]; then
    read -p "kafka[${version}] is installed, reinstall it? (y/N) " reinstall </dev/tty
    reinstall=$(echo "$reinstall" | tr -s '[:space:]')
    if [[ "$reinstall" != [Yy] ]]; then exit 0; fi
fi
$SUDO rm -rf /opt/kafka/${version}/*
$SUDO rm -rf /etc/systemd/system/kafka-${version}.service
$SUDO rm -rf kafka_*.tgz

ps -ef | grep kafkaServer | grep -v grep
if [[ $? == 0 ]]; then echo -e "\n[ERROR] please stop kafka service at first."; exit -1; fi

if [[ -z "$JAVA_HOME" ]]; then echo "please set JAVA_HOME for kafka."; exit -1; fi

read -p "Enter kafka[${version}] listen host (default localhost): " kafka_listen_host </dev/tty
kafka_listen_host=$(echo "$kafka_listen_host" | tr -s '[:space:]')
if [[ "$kafka_listen_host" == "" ]]; then kafka_listen_host="localhost"; fi

read -p "Enter kafka[${version}] listen port (default 9092): " kafka_listen_port </dev/tty
kafka_listen_port=$(echo "$kafka_listen_port" | tr -s '[:space:]')
if [[ "$kafka_listen_port" == "" ]]; then kafka_listen_port="9092"; fi

read -p "Enter kafka[${version}] controller port (default 9093): " kafka_controller_port </dev/tty
kafka_controller_port=$(echo "$kafka_controller_port" | tr -s '[:space:]')
if [[ "$kafka_controller_port" == "" ]]; then kafka_controller_port="9093"; fi

read -p "Enter kafka[${version}] user[admin] password (default admin): " kafka_admin_password </dev/tty
kafka_admin_password=$(echo "$kafka_admin_password" | tr -s '[:space:]')
if [[ "$kafka_admin_password" == "" ]]; then kafka_admin_password="admin"; fi

read -p "Enter kafka ui manager listen port (default 8080): " kafka_ui_port </dev/tty
kafka_ui_port=$(echo "$kafka_ui_port" | tr -s '[:space:]')
if [[ "$kafka_ui_port" == "" ]]; then kafka_ui_port="8080"; fi

wget https://mirrors.huaweicloud.com/apache/kafka/${version}/kafka_2.13-${version}.tgz
$SUDO tar zxvf kafka_2.13-${version}.tgz -C /opt/kafka/${version} --strip-components=1
rm -f kafka_2.13-${version}.tgz

wget https://gh.xmly.dev/https://github.com/provectus/kafka-ui/releases/download/v0.7.2/kafka-ui-api-v0.7.2.jar -O kafka-ui.jar
$SUDO mv kafka-ui.jar /opt/kafka/${version}/bin/kafka-ui.jar

$SUDO mkdir -p /opt/kafka/${version}/data

$SUDO sed 's/:9092/:'${kafka_listen_port}'/' -i /opt/kafka/${version}/config/kraft/server.properties
$SUDO sed 's/:9093/:'${kafka_controller_port}'/' -i /opt/kafka/${version}/config/kraft/server.properties
$SUDO sed 's#^log.dirs=.*#log.dirs=/opt/kafka/'${version}'/data#' -i /opt/kafka/${version}/config/kraft/server.properties
$SUDO sed 's/^listeners=PLAINTEXT/listeners=SASL_PLAINTEXT/' -i /opt/kafka/${version}/config/kraft/server.properties
$SUDO sed 's/^inter.broker.listener.name=PLAINTEXT/inter.broker.listener.name=SASL_PLAINTEXT/' -i /opt/kafka/${version}/config/kraft/server.properties
$SUDO sed 's#^advertised.listeners=PLAINTEXT://localhost#advertised.listeners=SASL_PLAINTEXT://'${kafka_listen_host}'#' -i /opt/kafka/${version}/config/kraft/server.properties
$SUDO sed 's/^listener.security.protocol.map=CONTROLLER:PLAINTEXT/listener.security.protocol.map=CONTROLLER:SASL_PLAINTEXT/' -i /opt/kafka/${version}/config/kraft/server.properties
cat << EOF | $SUDO tee -a /opt/kafka/${version}/config/kraft/server.properties
sasl.mechanism.controller.protocol=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
sasl.mechanism=PLAIN
auto.create.topics.enable=false
allow.everyone.if.no.acl.found=false
super.users=User:admin
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
EOF

cat << EOF | $SUDO tee /opt/kafka/${version}/config/kraft/server_sasl.conf
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="${kafka_admin_password}"
    user_admin="${kafka_admin_password}";
};
EOF

cat << EOF | $SUDO tee /opt/kafka/${version}/config/kraft/client_sasl.conf
KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="${kafka_admin_password}";
};
EOF

cat << EOF | $SUDO tee /opt/kafka/${version}/config/kraft/jaas.properties
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required  username="admin"  password="${kafka_admin_password}";
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
EOF

cat << EOF | $SUDO tee /opt/kafka/${version}/config/kafka-ui.yml
# 服务端设置
server:
  port: ${kafka_ui_port}
  servlet:
    context-path: /

# 权限设置
auth:
  type: LOGIN_FORM
spring:
  security:
    user:
      name: admin
      password: ${kafka_admin_password}

# kafka设置
dynamic:
  config:
    enabled: false
kafka:
  clusters:
    - bootstrapServers: ${kafka_listen_host}:${kafka_listen_port}
      name: local
      readOnly: false
      properties:
        security:
          protocol: "SASL_PLAINTEXT"
        sasl:
          mechanism: "PLAIN"
          jaas:
            config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"${kafka_admin_password}\";"
EOF

$SUDO chown -R ${USER} /opt/kafka/${version}
/opt/kafka/${version}/bin/kafka-server-start.sh --version
if [[ $? != 0 ]]; then echo -e "\n[ERROR] please resolve kafka deps with '/opt/kafka/${version}/bin/kafka-server-start.sh --version'."; exit -1; fi

CLUSTER_ID=$(/opt/kafka/${version}/bin/kafka-storage.sh random-uuid)
/opt/kafka/${version}/bin/kafka-storage.sh format --cluster-id ${CLUSTER_ID} --config /opt/kafka/${version}/config/kraft/server.properties

cat << EOF | $SUDO tee /etc/systemd/system/kafka-${version}.service
[Unit]
Description=Kafka Service - ${version}

[Service]
Restart=on-failure
RestartSec=30s
User=${USER}
ExecStart=/opt/kafka/${version}/bin/kafka-server-start.sh /opt/kafka/${version}/config/kraft/server.properties
Environment="JAVA_HOME=${JAVA_HOME}"
Environment="KAFKA_HEAP_OPTS=-Xmx1G -Xms1G -Djava.security.auth.login.config=/opt/kafka/${version}/config/kraft/server_sasl.conf"

[Install]
WantedBy=multi-user.target
Alias=kafka-${version}.service
EOF

cat << EOF | $SUDO tee /etc/systemd/system/kafka-ui.service
[Unit]
Description=Kafka UI Service

[Service]
Restart=on-failure
RestartSec=30s
User=${USER}
ExecStart=${JAVA_HOME}/bin/java -Xmx1G -Xms1G -Dsun.jnu.encoding=UTF-8 -Dfile.encoding=UTF-8 -Duser.timezone=GMT+08 --add-opens java.rmi/javax.rmi.ssl=ALL-UNNAMED -jar /opt/kafka/${version}/bin/kafka-ui.jar --spring.config.additional-location=/opt/kafka/${version}/config/kafka-ui.yml

[Install]
WantedBy=multi-user.target
Alias=kafka-ui.service
EOF

echo -e "\nMake sure that the following content exists in the \$HOME/.bashrc"
echo -e "\texport KAFKA_HOME=/opt/kafka/${version}"
echo -e "\texport PATH=\$KAFKA_HOME/bin:\$PATH"
echo -e "\nstart kafka service => "
echo -e "\n\tsudo systemctl start kafka-${version}"
echo -e "\tsudo systemctl start kafka-ui"
echo -e "\nexamples:"
echo -e "\n\tcreate topic => "
echo -e "\n\t/opt/kafka/${version}/bin/kafka-topics.sh --bootstrap-server ${kafka_listen_host}:${kafka_listen_port} --create --topic testTopic --partitions 1 --replication-factor 1  --command-config /opt/kafka/${version}/config/kraft/jaas.properties"
echo -e "\n\tproduce into topic => "
echo -e "\n\t/opt/kafka/${version}/bin/kafka-console-producer.sh broker-list --bootstrap-server ${kafka_listen_host}:${kafka_listen_port} --topic testTopic --producer.config /opt/kafka/${version}/config/kraft/jaas.properties"
echo -e "\n\tconsume topic => "
echo -e "\n\t/opt/kafka/${version}/bin/kafka-console-consumer.sh --bootstrap-server ${kafka_listen_host}:${kafka_listen_port} --topic testTopic --from-beginning --consumer.config /opt/kafka/${version}/config/kraft/jaas.properties"
echo -e "\nkafka ui:"
echo -e "\n\thttp://${kafka_listen_host}:${kafka_ui_port} (admin/${kafka_admin_password})"